package WindowAndWatermark

import Source.SensorReading
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object WindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    //每十五秒统计窗口内个传感器的温度最小值,以及最新的时间戳
    val resultStream = dataStream
      .map(data => (data.id, data.temperature, data.timeStamp))
      .keyBy(_._1) //按照二元组第一个元素id分组
      //      .countWindow(10)
      //      .window(EventTimeSessionWindows.withGap(Time.seconds(15)))
      // .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(3))) //滑动时间窗口
      //      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .timeWindow(Time.seconds(5))
      .reduce((CurRes, newData) => (CurRes._1, CurRes._2.min(newData._2), newData._3))

    resultStream.print("res")

    env.execute()
  }
}
